package com.fanghuaiming.test.c4.多路复用.多线程优化;

import static com.fanghuaiming.netty.utils.ByteBufferUtil.debugAll;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;
import lombok.extern.slf4j.Slf4j;

/****
 * @description:
 * @version:1.0.0
 * @author fanghuaiming
 * @data Created in 2021/9/25 下午2:58
 *
 */
@Slf4j
public class MultiThreadServer {

  public static void main(String[] args) throws IOException {
      Thread.currentThread().setName("boss");
      ServerSocketChannel ssc = ServerSocketChannel.open();
      ssc.configureBlocking(false);
      Selector boss = Selector.open();
      SelectionKey bosskey = ssc.register(boss, SelectionKey.OP_ACCEPT, null);
      ssc.bind(new InetSocketAddress(8080));
      //1.创建固定数量的worker
      Worker[] workers = new Worker[Runtime.getRuntime().availableProcessors()];
      //Runtime.getRuntime().availableProcessors()在Docker中拿到的是物理机器的cpu个数,而不是Docker分配的cpu个数
      for (int i = 0; i < workers.length; i++) {
        workers[i] = new Worker("worker-"+i);
      }
      AtomicInteger index = new AtomicInteger();
      while (true){
          boss.select();
          Iterator<SelectionKey> iterator = boss.selectedKeys().iterator();
          while (iterator.hasNext()){
              SelectionKey key = iterator.next();
              iterator.remove();
              if (key.isAcceptable()){
                  SocketChannel sc = ssc.accept();
                  sc.configureBlocking(false);
                  log.info("connected...:{}",sc.getRemoteAddress());
                  //2.关联 selector
                  log.info("before register...:{}",sc.getRemoteAddress());
                  //负载均衡算法去判断当前的channel应该注册到哪一个worker上
                  workers[index.getAndIncrement() % workers.length].register(sc);
                  log.info("after register...:{}",sc.getRemoteAddress());

              }
          }
      }
  }

  static class Worker implements Runnable{

      private Thread thread;
      private Selector selector;
      private String name;
      private volatile boolean start = false;
      private ConcurrentLinkedQueue<Runnable> queue = new ConcurrentLinkedQueue<>();

      public Worker(String name) {
          this.name = name;
      }

      //初始化线程和selector
      public void register(SocketChannel sc) throws IOException {
          if(!start){
              thread = new Thread(this,name);
              selector = Selector.open();
              thread.start();
              start = true;
          }
          //向队列添加任务,任务并没有被执行
          queue.add(() ->{
              try {
                  sc.register(selector,SelectionKey.OP_READ,null);
              } catch (ClosedChannelException e) {
                  e.printStackTrace();
              }
          });
          //因为下面的selector.select()先执行就已经阻塞住了
          selector.wakeup();//为了唤醒run中的selector.select();
          //wakeup类似于通行证一样的东西,无论先wakeup还是后wakeup都可以让select不阻塞继续运行
      }

      @Override
      public void run() {
          while (true){
              try {
                  selector.select();
                  Runnable task = queue.poll();
                  if(task != null){
                      task.run();//在这个位置执行了register
                  }
                  Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                  while (iterator.hasNext()){
                      SelectionKey key = iterator.next();
                      iterator.remove();
                      if(key.isReadable()){
                          ByteBuffer buffer = ByteBuffer.allocate(16);
                          SocketChannel channel = (SocketChannel) key.channel();
                          log.info("read...:{}",channel.getRemoteAddress());
                          channel.read(buffer);
                          buffer.flip();
                          debugAll(buffer);
                      }
                  }
              } catch (IOException e) {
                  e.printStackTrace();
              }
          }
      }
  }
}
